package aos.maekawa;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.PriorityQueue;
import java.util.Queue;

import aos.maekawa.entity.GlobalVariables;
import aos.maekawa.entity.Message;
import aos.maekawa.entity.MessageType;
import aos.maekawa.entity.ReportData;
import aos.maekawa.entity.RouteTable;
import aos.maekawa.entity.RoutineData;
import aos.maekawa.exception.AosException;
import aos.maekawa.log.LogUtils;
import aos.maekawa.socket.SocketHandler;
import aos.maekawa.socket.SocketListener;

public class ProcessorNode implements Runnable {
	private int nodeId;
	private int reportRecv;
	private int grantToId;
	private int reqId;
	private HashMap<Integer, Boolean> quorumGrant;
	private HashSet<Integer> quorumReject;
	private Object synObject;
	private RouteTable routeTable;
	private Queue<Message> messageQueue;
	private Queue<Integer> csRequestQueue;
	private Queue<Message> enquireQueue;
	private HashMap<Integer, SocketHandler> id2Socket;
	private SocketListener tcpServer;
	private ReportData report;

	public ProcessorNode(int port, String propertiesFile) throws AosException {
		reportRecv = 0;
		grantToId = -1;
		reqId = 0;
		routeTable = new RouteTable(propertiesFile);
		setNodeId(routeTable.getLocalNodeId());
		synObject = new Object();

		quorumGrant = new HashMap<Integer, Boolean>();
		for (Integer quorumNodeId : RoutineData.quorumList.get(nodeId)) {
			quorumGrant.put(quorumNodeId, false);
		}

		quorumReject = new HashSet<Integer>();

		messageQueue = new LinkedList<Message>();
		csRequestQueue = new PriorityQueue<Integer>();
		enquireQueue = new LinkedList<Message>();
		report = new ReportData(nodeId);

		new Thread(this).start();

		id2Socket = new HashMap<Integer, SocketHandler>();
		tcpServer = new SocketListener(port, id2Socket, routeTable,
				messageQueue, report);
		new Thread(tcpServer).start();

		if (nodeId == GlobalVariables.masterId) {
			MasterNode masterNode = new MasterNode(this);
			masterNode.startAll();
		}
	}

	@Override
	public void run() {
		while (true) {
			synchronized (messageQueue) {
				while (!messageQueue.isEmpty()) {
					Message msg = messageQueue.poll();
					try {
						handleMessageReceive(msg);
					} catch (AosException e) {
						e.printStackTrace();
					}
				}
				try {
					messageQueue.wait(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	private void handleMessageReceive(Message msg) throws AosException {
		LogUtils.debugLog("handleMessageReceive: " + msg.toString());
		switch (msg.getMsgType()) {
		case INIT:
			handleInit(msg);
			break;
		case REPORT:
			handleReport(msg);
			break;
		case END:
			handleEnd(msg);
			break;
		case REQUEST:
			handleRequest(msg);
			break;
		case REPLY:
			handleReply(msg);
			break;
		case RELEASE:
			handleRelease(msg);
			break;
		case FAIL:
			handleFail(msg);
			break;
		case ENQUIRE:
			handleEnquire(msg);
			break;
		case YIELD:
			handleYield(msg);
			break;
		case CS:
			handleCS(msg);
			break;
		case ACK:
			handleAck(msg);
			break;
		default:
		}
		LogUtils.debugLog("quorumGrant status");
		for (Integer nodeId : quorumGrant.keySet()) {
			LogUtils.debugLog(nodeId + " " + quorumGrant.get(nodeId));
		}
		LogUtils.debugLog("quorumReject status");
		for (Integer nodeId : quorumReject) {
			LogUtils.debugLog(nodeId + "");
		}
	}

	private void handleReply(Message msg) {
		quorumGrant.put(msg.getFromId(), true);
		quorumReject.remove(msg.getFromId());
		boolean grant = true;
		for (Boolean g : quorumGrant.values()) {
			if (!g) {
				grant = false;
				break;
			}
		}
		if (grant) {
			assert (quorumReject.isEmpty());
			Message csMsg = new Message();
			csMsg.setMsgType(MessageType.CS);
			csMsg.setFromId(nodeId);
			csMsg.setToId(GlobalVariables.csNodeId);
			csMsg.setCst(RoutineData.cst);
			csMsg.setReqId(reqId);
			reqId++;
			try {
				sendMessage(csMsg);
			} catch (AosException e) {
				e.printStackTrace();
			}
		}
	}

	private void handleFail(Message msg) {
		quorumGrant.put(msg.getFromId(), false);
		quorumReject.add(msg.getFromId());
		while (!enquireQueue.isEmpty()) {
			handleEnquire(enquireQueue.poll());
		}
	}

	private void handleAck(Message msg) {
		assert (quorumReject.isEmpty());
		enquireQueue.clear();
		for (Integer toNodeId : quorumGrant.keySet()) {
			quorumGrant.put(toNodeId, false);
			Message newMsg = new Message();
			newMsg.setFromId(nodeId);
			newMsg.setToId(toNodeId);
			newMsg.setMsgType(MessageType.RELEASE);
			try {
				sendMessage(newMsg);
			} catch (AosException e) {
				e.printStackTrace();
			}
		}
		synchronized (synObject) {
			synObject.notify();
			LogUtils.debugLog("synObject notify");
		}
	}

	private void handleRequest(Message msg) {
		Message newMsg = new Message();
		if (grantToId == -1) {
			newMsg.setFromId(nodeId);
			newMsg.setToId(msg.getFromId());
			newMsg.setMsgType(MessageType.REPLY);
			grantToId = msg.getFromId();
		} else if (msg.getFromId() < grantToId) {
			csRequestQueue.add(msg.getFromId());
			newMsg.setFromId(nodeId);
			newMsg.setToId(grantToId);
			newMsg.setMsgType(MessageType.ENQUIRE);
		} else {
			csRequestQueue.add(msg.getFromId());
			newMsg.setFromId(nodeId);
			newMsg.setToId(msg.getFromId());
			newMsg.setMsgType(MessageType.FAIL);
		}
		try {
			sendMessage(newMsg);
		} catch (AosException e) {
			e.printStackTrace();
		}
	}

	private void handleRelease(Message msg) {
		grantToId = -1;
		if (!csRequestQueue.isEmpty()) {
			int toNodeId = csRequestQueue.poll();
			Message newMsg = new Message();
			newMsg.setFromId(nodeId);
			newMsg.setToId(toNodeId);
			newMsg.setMsgType(MessageType.REPLY);
			grantToId = toNodeId;
			try {
				sendMessage(newMsg);
			} catch (AosException e) {
				e.printStackTrace();
			}
		}
	}

	private void handleYield(Message msg) {
		int toNodeId = csRequestQueue.poll();
		csRequestQueue.add(msg.getFromId());
		Message newMsg = new Message();
		newMsg.setFromId(nodeId);
		newMsg.setToId(toNodeId);
		newMsg.setMsgType(MessageType.REPLY);
		grantToId = toNodeId;
		try {
			sendMessage(newMsg);
		} catch (AosException e) {
			e.printStackTrace();
		}
	}

	private void handleEnquire(Message msg) {
		if (!quorumReject.isEmpty()) {
			Message respMsg = new Message();
			respMsg.setFromId(nodeId);
			respMsg.setToId(msg.getFromId());
			respMsg.setMsgType(MessageType.YIELD);
			quorumGrant.put(msg.getFromId(), false);
			quorumReject.add(msg.getFromId());
			try {
				sendMessage(respMsg);
			} catch (AosException e) {
				e.printStackTrace();
			}
		} else {
			enquireQueue.add(msg);
		}
	}

	private void handleInit(Message msg) {
		CSRoutine csRoutine = new CSRoutine(this, synObject);
		new Thread(csRoutine).start();
	}

	private void handleCS(Message msg) {
		CSHandler csHandler = new CSHandler(this, msg);
		new Thread(csHandler).start();
	}

	private void handleReport(Message msg) {
		LogUtils.keyLog(msg.getReport().toString());
		msg.getReport().dump();
		++reportRecv;
		int n = RoutineData.n;
		if (reportRecv == n) {
			for (int i = 0; i < n; ++i) {
				Message endMsg = new Message();
				endMsg.setMsgType(MessageType.END);
				endMsg.setFromId(nodeId);
				endMsg.setToId(i);
				try {
					sendMessage(endMsg);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

	}

	private void handleEnd(Message msg) {
		LogUtils.runLog("END");
	}

	public synchronized void sendMessage(Message msg, int port)
			throws AosException {
		SocketHandler socketHandler = null;
		socketHandler = id2Socket.get(msg.getToId());
		if (socketHandler == null) {
			Socket socket = null;
			try {
				socket = new Socket(routeTable.getIpAddress(msg.getToId()),
						port);
			} catch (IOException e) {
				e.printStackTrace();
				return;
			}
			socketHandler = new SocketHandler(socket, messageQueue, report);
			id2Socket.put(msg.getToId(), socketHandler);
			new Thread(socketHandler).start();
		}
		socketHandler.sendMessage(msg);
		LogUtils.debugLog("quorumGrant status");
		for (Integer nodeId : quorumGrant.keySet()) {
			LogUtils.debugLog(nodeId + " " + quorumGrant.get(nodeId));
		}
		LogUtils.debugLog("quorumReject status");
		for (Integer nodeId : quorumReject) {
			LogUtils.debugLog(nodeId + "");
		}
	}

	public void sendMessage(Message msg) throws AosException {
		sendMessage(msg, GlobalVariables.port);
	}

	public int getNodeId() {
		return nodeId;
	}

	public void setNodeId(int nodeId) {
		this.nodeId = nodeId;
	}

	public ReportData getReport() {
		return report;
	}

	public static void main(String args[]) throws Exception {
		try {
			GlobalVariables.initialize("global.txt");
			RoutineData.initialize(GlobalVariables.input);
		} catch (Exception e) {
			e.printStackTrace();
		}
		LogUtils.initialize(GlobalVariables.loglevel, InetAddress
				.getLocalHost().getHostName() + "_" + GlobalVariables.logfile);

		ProcessorNode node = new ProcessorNode(GlobalVariables.port,
				GlobalVariables.iptable);
		new Thread(node).start();
	}

}
